package com.zhanggm.zyk.email.kafka;


import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.zhanggm.zyk.email.utils.MailUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {

    @Autowired
    private MailUtil mailUtil;

    @Autowired
    RedissonClient redissonClient;




    @KafkaListener(topics = {"email0308"})
    public void cunsumer(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
        //消息内容
        final String msgJson = consumerRecord.value();
        String key = consumerRecord.key();
        JSONObject msgJsonObj = JSONUtil.parseObj(msgJson);
        String email = msgJsonObj.getStr("email");

        final RLock lock = redissonClient.getLock(key);
        try {
            if (lock.tryLock()) {
                mailUtil.sendSimpleMail(email, "合同的审核通知", msgJsonObj.get("name") + "审核通过了");
                acknowledgment.acknowledge();
            }
        } catch (Exception e) {
            e.printStackTrace();
            acknowledgment.nack(1000);
        }finally {
            lock.unlock();
        }


    }

}
